package com.ldj.rocketmq.consumer;

import com.alibaba.fastjson.JSON;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;

/**
 * User: ldj
 * Date: 2024/5/26
 * Time: 16:08
 * Description: 顺序消息消费者
 */
public class OrderConsumer {

    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer-group-order");
        consumer.setNamesrvAddr("192.168.208.190:9876;192.168.208.191:9876;192.168.208.192:9876");
        consumer.subscribe("OrderTopic", "*");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.setMessageModel(MessageModel.CLUSTERING);

        //订单id做为key,消息作为value
        Map<String, Set<String>> concurrentHashMap = new ConcurrentHashMap<>();
        consumer.registerMessageListener((List<MessageExt> msgs, ConsumeConcurrentlyContext context) -> {
            msgs.forEach(msg -> {
                String str = new String(msg.getBody());
                int left = str.indexOf("[");
                int right = str.indexOf("]");
                String substring = str.substring(left + 1, right);
                String[] k_v = substring.split("-");
                Set<String> messages = concurrentHashMap.get(k_v[0]);
                if (CollectionUtils.isEmpty(messages)) {
                    Set<String> msgFirstList = new CopyOnWriteArraySet<>();
                    msgFirstList.add(str);
                    concurrentHashMap.put(k_v[0], msgFirstList);
                } else {
                    Set<String> msgAddList = concurrentHashMap.get(k_v[0]);
                    msgAddList.add(str);
                    concurrentHashMap.put(k_v[0], msgAddList);
                }
            });

            System.out.println(JSON.toJSONString(concurrentHashMap));
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
        System.out.println("消费者准备就绪...");
    }
}
